package com.yzh.luckydraw.application.worker;

import com.alibaba.fastjson.JSON;
import com.xxl.job.core.context.XxlJobHelper;
import com.xxl.job.core.handler.annotation.XxlJob;
import com.yzh.dbrouter.strategy.IDBRouterStrategy;
import com.yzh.luckydraw.application.mq.producer.KafkaProducer;
import com.yzh.luckydraw.common.Constants;
import com.yzh.luckydraw.common.Result;
import com.yzh.luckydraw.domain.activity.model.vo.ActivityVO;
import com.yzh.luckydraw.domain.activity.model.vo.InvoiceVO;
import com.yzh.luckydraw.domain.activity.service.deploy.IActivityDeploy;
import com.yzh.luckydraw.domain.activity.service.partake.IActivityPartake;
import com.yzh.luckydraw.domain.activity.service.stateflow.IStateHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;
import java.util.Date;
import java.util.List;

/**
 * 抽奖业务，任务配置
 */
@Component
public class LuckyDrawXxlJob {
    private Logger logger = LoggerFactory.getLogger(LuckyDrawXxlJob.class);
    @Resource
    private IActivityDeploy activityDeploy;
    @Resource
    private IActivityPartake activityPartake;
    @Resource
    private IStateHandler stateHandler;
    @Resource
    private IDBRouterStrategy dbRouter;
    @Resource
    private KafkaProducer kafkaProducer;

    @XxlJob("luckyDrawActivityStateJobHandler")
    public void luckyDrawActivityStateJobHandler() {
        logger.info("扫描活动状态：开始");
        List<ActivityVO> activityVOList = activityDeploy.scanToDoActivityList(0L);
        if (activityVOList.isEmpty()) {
            logger.info("扫描活动状态：结束-暂无符合需要扫描的活动");
            return;
        }
        while (!activityVOList.isEmpty()) {
            for (ActivityVO activityVO : activityVOList) {
                Integer state = activityVO.getState();
                switch (state) {
                    case 4:
                        Result doingresult = stateHandler.doing(activityVO.getActivityId(), Constants.ActivityState.PASS);
                        logger.info("扫描活动状态为活动中 结果：{} activityId：{} activityName：{} creator：{}", JSON.toJSONString(doingresult), activityVO.getActivityId(), activityVO.getActivityName(), activityVO.getCreator());
                        break;
                    case 5:
                        if (activityVO.getEndDateTime().before(new Date())) {
                            Result closeResult = stateHandler.close(activityVO.getActivityId(), Constants.ActivityState.DOING);
                            logger.info("扫描活动状态为关闭 结果：{} activityId：{} activityName：{} creator：{}", JSON.toJSONString(closeResult), activityVO.getActivityId(), activityVO.getActivityName(), activityVO.getCreator());
                        }
                        break;
                    default:
                        break;
                }
            }
            //拿到最后一个记录，从该记录继续向后扫描10条记录
            ActivityVO activityVO = activityVOList.get(activityVOList.size() - 1);
            activityVOList = activityDeploy.scanToDoActivityList(activityVO.getId());
        }
        logger.info("扫描活动状态：结束");
    }

    @XxlJob("luckyDrawOrderMQStateJobHandler")
    public void luckyDrawOrderMQStateJobHandler() {
        //验证参数，从xxl前端界面传参
        String jobParam = XxlJobHelper.getJobParam();
        if (null == jobParam) {
            logger.info("扫描用户奖品发放MQ状态：错误-参数为null");
            return;
        }

        //获取分布式任务配置参数 参数格式1,2 也可以指定一个。也可以配置多个库，按照部署的集群进行数量配置，分别扫描效率更高
        String[] params = jobParam.split(",");
        logger.info("扫描用户奖品发放MQ状态：开始。params：{}", JSON.toJSONString(params));

        if (params.length == 0) {
            logger.info("扫描用户奖品发放MQ状态：错误-参数为null");
            return;
        }
        //获取分库分表的分表数量
        int tbCount = dbRouter.tbCount();

        //循环扫描库
        for (String param : params) {
            int db = Integer.parseInt(param);
            //判断库是否存在
            if (db > dbRouter.dbCount()) {
                logger.info("扫描用户奖品发放MQ状态：库不存在");
                continue;
            }

            //循环扫描库中的所有表
            for (int i = 0; i < tbCount; i++) {
                //扫描库表数据
                List<InvoiceVO> invoiceVOList = activityPartake.scanInvoiceMqState(db, i);
                logger.info("扫描用户奖品发放MQ状态：库：{} 表：{} 扫描数量：{}", db, i, invoiceVOList.size());

                //补偿MQ
                for (InvoiceVO invoiceVO : invoiceVOList) {
                    ListenableFuture<SendResult<String, Object>> future = kafkaProducer.sendInvoice(invoiceVO);
                    future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
                        @Override
                        public void onSuccess(SendResult<String, Object> stringObjectSendResult) {
                            //MQ消息发送完成，更新数据库表 user_strategy_export.mq_state = 1
                            activityPartake.updateInvoiceMqState(invoiceVO.getuId(), invoiceVO.getOrderId(), Constants.MQState.COMPLETE.getCode());
                        }
                        @Override
                        public void onFailure(Throwable throwable) {
                            // MQ 消息发送失败，更新数据库表 user_strategy_export.mq_state = 2 【等待定时任务扫码补偿MQ消息】
                            activityPartake.updateInvoiceMqState(invoiceVO.getuId(), invoiceVO.getOrderId(), Constants.MQState.FAIL.getCode());
                        }
                    });
                }
            }
        }
        logger.info("扫描用户奖品发放MQ状态：完成 params：{}", JSON.toJSONString(params));
    }

}
